package com.lmm.rpc.core.client;

import com.lmm.rpc.core.client.async.RpcCallback;
import com.lmm.rpc.core.client.protocol.MessageProtocol;
import com.lmm.rpc.core.client.protocol.RequestMetadata;
import com.lmm.rpc.core.cluster.RemoteServerInfo;
import com.lmm.rpc.core.codec.DefaultDecoder;
import com.lmm.rpc.core.codec.DefaultEncoder;
import com.lmm.rpc.core.common.reponse.DefaultRepose;
import com.lmm.rpc.core.common.request.DefaultRequest;
import com.lmm.rpc.core.exception.RpcException;
import com.lmm.rpc.core.handler.DefaultResponseHandler;
import com.lmm.rpc.core.transport.LocalRpcResponseCache;
import com.lmm.rpc.core.transport.RpcFuture;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * netty 客户端
 * */
@Slf4j
@Data
public class NettyClient implements  NettyClientTransport<Channel>{
    private static final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();
    private final Bootstrap bootstrap;
    private final DefaultResponseHandler responseHandler = new DefaultResponseHandler();
    private final Timer timer = new Timer("RpcClientHouseKeepingService", true);
    private ThreadPoolExecutor callBackExecutor;
    private ThreadPoolExecutor defaultAsyncSenderExecutor;

    public Bootstrap getBootstrap() {
        return bootstrap;
    }

    NettyClient(){
        this.bootstrap = new Bootstrap();
        this.open();
        this.defaultAsyncSenderExecutor = new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors(),
                Runtime.getRuntime().availableProcessors(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(),
                new ThreadFactory() {
                    private final AtomicInteger threadIndex = new AtomicInteger(0);

                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "AsyncRpcSenderExecutor_" + this.threadIndex.incrementAndGet());
                    }
                });
    }

    //初始化客户端
    private boolean open(){
        //初始化客户端    设置bootstrap的基本信息
        bootstrap.group(nioEventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //业务处理器
                        socketChannel.pipeline()
                                //解码
                                .addLast(new DefaultDecoder())
                                .addLast(responseHandler)
                                //编码
                                .addLast(new DefaultEncoder<>());
                    }
                });
        this.timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                try {
                    NettyClient.this.scanResponseTable();
                } catch (Throwable e) {
                    log.error("scanResponseTable exception", e);
                }
            }
        }, 1000 * 3, 1000);
        return true;
    }

    /**
     * 获取连接好的channel
     *
     * @param remoteServer*/
    public Channel getConnectChannel(RemoteServerInfo remoteServer){
        return connect(remoteServer);
    }

    private Channel connect(RemoteServerInfo remoteServer){
        //避免网络抖动的原因 尝试3次
        int tryTime = 3;
        Channel channel = null;
        while (tryTime-->0){

            try {
                //同步建立连接
                ChannelFuture channelFuture = bootstrap.connect(remoteServer.getServer(), Integer.parseInt(remoteServer.getPort())).sync();
                log.info("connect rpc server {} on port {} success.",remoteServer.getServer(), remoteServer.getPort());
                channel = channelFuture.channel();
                return channel;
            } catch (Exception e) {
                log.warn("connect rpc server {} on port {} faild. retry {} again",remoteServer.getServer(), remoteServer.getPort(),3-tryTime);
                e.printStackTrace();
            }
        }
        throw new RpcException("connect rpc server on port faild finally"+remoteServer.getServer()+":"+remoteServer.getPort());
    }

    @Override
    public MessageProtocol<DefaultRepose> sendRequest(RequestMetadata metadata,Channel channel) throws Exception {
        MessageProtocol<DefaultRequest> protocol = metadata.getProtocol();
        RpcFuture<MessageProtocol<DefaultRepose>> reposeFuture = new RpcFuture<>();
        reposeFuture.setTimeoutMillis(metadata.getTimeout());
        LocalRpcResponseCache.add(String.valueOf(protocol.getHeader().getRequestId()),reposeFuture);
        channel.writeAndFlush(protocol);
        log.info("send message success requestId:{}",protocol.getHeader().getRequestId());
        return metadata.getTimeout() != null ? reposeFuture.get(metadata.getTimeout(), TimeUnit.MILLISECONDS) : reposeFuture.get();
    }

    @Override
    public RpcFuture<MessageProtocol<DefaultRepose>> asyncSendRequest(RequestMetadata metadata, Channel channel, RpcCallback rpcCallback) {
        RpcFuture<MessageProtocol<DefaultRepose>> reposeFuture = new RpcFuture<>();
        reposeFuture.setTimeoutMillis(metadata.getTimeout());
        reposeFuture.setRpcCallback(rpcCallback);
        String requestId = String.valueOf(metadata.getProtocol().getHeader().getRequestId());
        LocalRpcResponseCache.getRequestResponseCache().put(requestId,reposeFuture);
        try {
            channel.writeAndFlush(metadata.getProtocol()).addListener((ChannelFutureListener) channelFuture -> {
                if (channelFuture.isSuccess()) {
                    //如果发送消息成功给Server，那么这里直接Set发出标记后return
                    reposeFuture.setSendRequestOK(true);
                    return;
                }
                //发送失败，做善后处理
                requestFail(requestId,channelFuture.cause());
            });
        }catch (Exception e){
            //发送失败，做善后处理
            requestFail(requestId,e);
        }
        return reposeFuture;
    }
    // 异步调用失败的处理逻辑
    private void requestFail(String requestId,Throwable e) {
        RpcFuture<MessageProtocol<DefaultRepose>> responseFuture = LocalRpcResponseCache.getRequestResponseCache().remove(requestId);
        if (Objects.nonNull(responseFuture)){
            responseFuture.setSendRequestOK(false);
            responseFuture.setResponse(null);
            //执行回调
            executeInvokeCallback(responseFuture,e);
        }
    }

    public void executeInvokeCallback(RpcFuture<MessageProtocol<DefaultRepose>> responseFuture,Throwable e) {
        ExecutorService executor = getCallBackExecutor();
        if (Objects.nonNull(executor)){
            executor.submit(() -> responseFuture.executeInvokeCallback(e));
        }
    }
    /**
     * 扫描响应table
     */
    public void scanResponseTable() {
        final List<RpcFuture<MessageProtocol<DefaultRepose>>> rfList = new LinkedList<>();
        Iterator<Map.Entry<String, RpcFuture<MessageProtocol<DefaultRepose>>>> it = LocalRpcResponseCache.getRequestResponseCache().entrySet().iterator();
        while (it.hasNext()) {
            RpcFuture<MessageProtocol<DefaultRepose>> rep = it.next().getValue();
            if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
                it.remove();
                rfList.add(rep);
                log.warn("remove timeout request, " + rep);
            }
        }
        for (RpcFuture<MessageProtocol<DefaultRepose>> rf : rfList) {
            try {
                executeInvokeCallback(rf,new RpcException("timeout request"));
            } catch (Throwable e) {
                log.warn("scanResponseTable, operationComplete Exception", e);
            }
        }
    }
}
